package com.aidex.config;

import java.util.Date;
import java.util.List;

import javax.jms.*;

import com.aidex.common.utils.JsonHelper;
import com.aidex.common.utils.StringUtils;
import com.aidex.response.NlpCutResponse;
import com.aidex.risk.domain.syncMethod;
import com.aidex.risk.domain.transData;
import com.aidex.system.domain.SyncEmergencyFlight;
import com.aidex.system.domain.Waybill;
import com.aidex.system.service.*;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import com.aidex.system.service.impl.WaybillServiceImpl;

import com.ibm.jms.JMSBytesMessage;
import com.ibm.mq.jms.MQQueueConnectionFactory;
import com.ibm.msg.client.wmq.WMQConstants;

import lombok.AllArgsConstructor;

@Component
@AllArgsConstructor
public class InitMq implements ApplicationRunner {
    private static final Logger log = LoggerFactory.getLogger(InitMq.class);


    @Autowired
    private   WaybillService waybillService;
    @Autowired
    private WaybillDetailService waybillDetailService;
    @Autowired
    private   SyncEmergencyFlightService syncEmergencyFlightService;

    @Autowired
    private ApiLogService apiLogService;
    @Autowired
    private RunLogService runLogService;
    public void ReceverMsg() {



      /*  MQQueueConnectionFactory mqQueueConnectionFactory_cfps1 = new MQQueueConnectionFactory();
        Connection connection_cfps1 = null;*/
        try { MQQueueConnectionFactory mqQueueConnectionFactory = new MQQueueConnectionFactory();
            Connection connection = null;
            mqQueueConnectionFactory.setHostName("10.80.33.208");//总线输出的MQ服务器IP
            mqQueueConnectionFactory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
            mqQueueConnectionFactory.setCCSID(1208);
            mqQueueConnectionFactory.setChannel("ESB.ADMIN.SVRCONN");//总线输出的MQ通道
            mqQueueConnectionFactory.setPort(40001);//总线输出的MQ服务器端口
            mqQueueConnectionFactory.setQueueManager("ESB_QM");//总线输出的MQ队列管理器

            connection = mqQueueConnectionFactory.createConnection();

            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("flfj_txd_to_ersuo_queue");//分类分级-货邮安检、天信达报文交换MQ通道

            MessageConsumer syncConsumer = session.createConsumer(queue);
            syncConsumer.setMessageListener(new MessageListener() {//设置监听器
                @Override
                public void onMessage(Message message) {
                    try {
                        String request = null;
                        if (message instanceof JMSBytesMessage) {
                            log.info("接收到MQ数据JMSBytesMessage");
                            JMSBytesMessage jmsBytesMessage = (JMSBytesMessage) message;
                            try {
                                if (null != jmsBytesMessage) {
                                    Long lenght = 2048L;
                                    if (lenght != null && lenght > 0) {
                                        lenght = jmsBytesMessage.getBodyLength();
                                    }
                                    byte[] read = new byte[lenght.intValue()];
                                    jmsBytesMessage.readBytes(read);
                                    request = new String(read, "UTF-8");
                                    log.info("接收到MQ数据JMSBytesMessage：" + request);//接收到的消息
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (message instanceof TextMessage) {
                            log.info("接收到MQ数据TextMessage");
                            TextMessage text = (TextMessage) message;
                            request = text.getText();
                            log.info("接收到MQ数据TextMessage" + request);//接收
                        }
                        if(!request.equals(null) && !request.isEmpty()){
                            syncMethod<Object> sync = new syncMethod<>();
                            sync = JsonHelper.getInstance().readValue(request,sync.getClass());
                            if(sync.getMethod().equals("syncDeclaration")){
                                log.info("接收到MQ订单数据" + request);//接收
                                syncMethod<transData> data = new syncMethod<>();
                                data =(syncMethod<transData>) JSON.parseObject(request,new TypeReference<syncMethod<transData>>(){});
                                data.setFrom("CSIS");
                                waybillService.Save(data);
                                apiLogService.updateScuess("MQ运单信息");
                            }else if (sync.getMethod().equals("syncEmergencyFlight") || sync.getMethod().equals("syncEmergencyGuard")){
                                log.info("接收到MQ航班数据" + request);//接收
                                syncMethod<List<SyncEmergencyFlight>> data = new syncMethod<>();
                                data =(syncMethod<List<SyncEmergencyFlight>>) JSON.parseObject(request,new TypeReference<syncMethod<List<SyncEmergencyFlight>>>(){});
                                List<SyncEmergencyFlight> infos = data.getData();
                                for(SyncEmergencyFlight info:infos){
                                    if(info.getFlightTime()==null ){
                                        continue;
                                    }
                                    info.setId(null);
                                    info.setType(data.getMethod());
                                    SyncEmergencyFlight flight = new SyncEmergencyFlight();
                                    flight.setFlight(info.getFlight());
                                    flight.setType(data.getMethod());
                                    flight.setFlightTime(info.getFlightTime());
                                    SyncEmergencyFlight oldFlight = syncEmergencyFlightService.get(flight);
                                    Boolean updateWayBill = Boolean.FALSE;
                                    if (oldFlight == null) {
                                        updateWayBill = syncEmergencyFlightService.save(info);
                                    } else if (StringUtils.isNotBlank(info.getStartupDisable()) &&
                                            !info.getStartupDisable().equals(oldFlight.getStartupDisable())) {
                                        oldFlight.setFlightTime(info.getFlightTime());
                                        oldFlight.setStartupDisable(info.getStartupDisable());
                                        updateWayBill = syncEmergencyFlightService.save(oldFlight);
                                    }
                                    if (updateWayBill) {
                                        flight = new SyncEmergencyFlight();
                                        flight.setFlight(info.getFlight());
                                        flight.setFlightTime(info.getFlightTime());
                                        List<SyncEmergencyFlight> flightList = syncEmergencyFlightService.findList(flight);
                                        List<Waybill> waybills = waybillService.findListByFlight(info.getFlight(), info.getFlightTime());
                                        String isImportant = "0";
                                        if (flightList != null && flightList.stream().count() > 0) {
                                            for (SyncEmergencyFlight item : flightList) {
                                                if (item.getStartupDisable() == "1") {
                                                    isImportant = "1";
                                                    break;
                                                }
                                            }
                                        }
                                        for (Waybill waybill : waybills) {
                                            if(!waybill.getDataSource().equalsIgnoreCase("ALMS")){
                                                continue;
                                            }
                                            if (isImportant != waybill.getIsImportantFlights()) {
                                                waybill.setStatus(0);
                                                waybill.setSendStatus(0);
                                                waybill.setIsImportantFlights("");
                                                waybill.setAgentLevel("");
                                                waybill.setRiskClassification("");
                                                waybill.setRiskRating("");
                                                waybillService.save(waybill);
                                                waybillDetailService.deleteWaybillDetailByWaybillId(waybill.getId());
                                            }
                                        }
                                    }
                                }
                                apiLogService.updateScuess("MQ航班信息");
                            }
                        }
                        //apiLogService.updateScuess("MQ");
                    } catch (JMSException e) {
                        apiLogService.updateError("MQ");
                        runLogService.saveError("MQ异常","发生错误JMSException，请查看日志记录");
                        log.error("10.80.33.208MQ ReceverMsg JMSException syncTradeConsumer"+e.getMessage());
                        //throw new RuntimeException("call ibm mq jms in error,", e);
                    } catch (JsonMappingException e) {
                        apiLogService.updateError("MQ");
                        runLogService.saveError("MQ异常","发生错误JsonMappingException，请查看日志记录");
                        log.error("10.80.33.208MQ ReceverMsg JsonMappingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (JsonProcessingException e) {
                        apiLogService.updateError("MQ");
                        runLogService.saveError("MQ异常","发生错误JsonProcessingException，请查看日志记录");
                        log.error("10.80.33.208MQ ReceverMsg JsonProcessingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (Throwable e){
                        apiLogService.updateError("MQ");
                        runLogService.saveError("MQ异常","发生错误，请查看日志记录");
                        log.error("10.80.33.208MQ ReceverMsg Throwable:"+e.getMessage());
                        System.out.println("MQ ReceverMsg Throwable:"+e.getMessage());
                    }
                }
            });
            connection.start();
        } catch (Exception e) {
//            apiLogService.updateError("MQ");
            e.printStackTrace();
            System.out.println(e.getMessage());
            log.error("10.80.33.208 MQ"+e.getMessage());
        }

        try{
            MQQueueConnectionFactory mqQueueConnectionFactory_cfps = new MQQueueConnectionFactory();
            Connection connection_cfps = null;
            mqQueueConnectionFactory_cfps.setHostName("10.71.248.34");//总线输出的MQ服务器IP
            mqQueueConnectionFactory_cfps.setTransportType(WMQConstants.WMQ_CM_CLIENT);
            mqQueueConnectionFactory_cfps.setCCSID(1208);
            mqQueueConnectionFactory_cfps.setChannel("ESB.ADMIN.SVRCONN");//总线输出的MQ通道
            mqQueueConnectionFactory_cfps.setPort(40001);//总线输出的MQ服务器端口
            mqQueueConnectionFactory_cfps.setQueueManager("ESB_QM");//总线输出的MQ队列管理器

            connection_cfps = mqQueueConnectionFactory_cfps.createConnection();

            Session session_cfps = connection_cfps.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            Queue queue_cfps = session_cfps.createQueue("flfj_txd_status_mq_topic");//分类分级-货邮安检、天信达报文交换MQ通道

            MessageConsumer syncConsumer_cfps = session_cfps.createConsumer(queue_cfps);
            syncConsumer_cfps.setMessageListener(new MessageListener() {//设置监听器
                @Override
                public void onMessage(Message message) {
                    try {
                        String request = null;
                        if (message instanceof JMSBytesMessage) {
                            log.info("接收到CFPS_MQ数据JMSBytesMessage");
                            JMSBytesMessage jmsBytesMessage = (JMSBytesMessage) message;
                            try {
                                if (null != jmsBytesMessage) {
                                    Long lenght = 2048L;
                                    if (lenght != null && lenght > 0) {
                                        lenght = jmsBytesMessage.getBodyLength();
                                    }
                                    byte[] read = new byte[lenght.intValue()];
                                    jmsBytesMessage.readBytes(read);
                                    request = new String(read, "UTF-8");
                                    log.info("接收到CFPS_MQ数据JMSBytesMessage：" + request);//接收到的消息
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (message instanceof TextMessage) {
                            log.info("接收到CFPS_MQ数据TextMessage");
                            TextMessage text = (TextMessage) message;
                            request = text.getText();
                            log.info("接收到CFPS_MQ数据TextMessage" + request);//接收
                        }
                        if(!request.equals(null) && !request.isEmpty()){
                            syncMethod<Object> sync = new syncMethod<>();
                            sync = JsonHelper.getInstance().readValue(request,sync.getClass());
                            if(sync.getMethod().equals("syncDeclarationInfo")){
                                log.info("接收到CFPS_MQ订单数据" + request);//接收
                                syncMethod<transData> data = new syncMethod<>();
                                data =(syncMethod<transData>) JSON.parseObject(request,new TypeReference<syncMethod<transData>>(){});
                                data.setFrom("CFPS");
                                waybillService.Save(data);
                                apiLogService.updateScuess("CFPS_MQ运单信息");
                            }
                        }
                        //apiLogService.updateScuess("CFPS_MQ");
                    } catch (JMSException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JMSException，请查看日志记录");
                        log.error("10.71.248.34CFPS_MQ ReceverMsg JMSException syncTradeConsumer"+e.getMessage());
                        //throw new RuntimeException("call ibm mq jms in error,", e);
                    } catch (JsonMappingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonMappingException，请查看日志记录");
                        log.error("10.71.248.34CFPS_MQ ReceverMsg JsonMappingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (JsonProcessingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonProcessingException，请查看日志记录");
                        log.error("10.71.248.34CFPS_MQ ReceverMsg JsonProcessingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (Throwable e){
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误，请查看日志记录");
                        log.error("10.71.248.34CFPS_MQ ReceverMsg Throwable:"+e.getMessage());
                        System.out.println("10.71.248.34CFPS_MQ ReceverMsg Throwable:"+e.getMessage());
                    }
                }
            });
            connection_cfps.start();
            while (true) {
            }
           /* mqQueueConnectionFactory_cfps1.setHostName("10.71.248.34");//总线输出的MQ服务器IP
            mqQueueConnectionFactory_cfps1.setTransportType(WMQConstants.WMQ_CM_CLIENT);
            mqQueueConnectionFactory_cfps1.setCCSID(1208);
            mqQueueConnectionFactory_cfps1.setChannel("ESB.ADMIN.SVRCONN");//总线输出的MQ通道
            mqQueueConnectionFactory_cfps1.setPort(40001);//总线输出的MQ服务器端口
            mqQueueConnectionFactory_cfps1.setQueueManager("ESB_QM");//总线输出的MQ队列管理器

            connection_cfps1 = mqQueueConnectionFactory_cfps1.createConnection();

            Session session_cfps1 = connection_cfps1.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            Queue queue_cfps1 = session_cfps1.createQueue("flfj_txd_status_mq_topic");//分类分级-货邮安检、天信达报文交换MQ通道

            MessageConsumer syncConsumer_cfps1 = session_cfps1.createConsumer(queue_cfps);
            syncConsumer_cfps1.setMessageListener(new MessageListener() {//设置监听器
                @Override
                public void onMessage(Message message) {
                    try {
                        String request = null;
                        if (message instanceof JMSBytesMessage) {
                            log.info("接收到CFPS_MQ数据JMSBytesMessage 1");
                            JMSBytesMessage jmsBytesMessage = (JMSBytesMessage) message;
                            try {
                                if (null != jmsBytesMessage) {
                                    Long lenght = 2048L;
                                    if (lenght != null && lenght > 0) {
                                        lenght = jmsBytesMessage.getBodyLength();
                                    }
                                    byte[] read = new byte[lenght.intValue()];
                                    jmsBytesMessage.readBytes(read);
                                    request = new String(read, "UTF-8");
                                    log.info("接收到CFPS_MQ数据JMSBytesMessage：" + request);//接收到的消息
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (message instanceof TextMessage) {
                            log.info("接收到CFPS_MQ数据TextMessage 1");
                            TextMessage text = (TextMessage) message;
                            request = text.getText();
                            log.info("接收到CFPS_MQ数据TextMessage 1" + request);//接收
                        }
                        if(!request.equals(null) && !request.isEmpty()){
                            syncMethod<Object> sync = new syncMethod<>();
                            sync = JsonHelper.getInstance().readValue(request,sync.getClass());
                            if(sync.getMethod().equals("syncDeclarationInfo")){
                                log.info("接收到CFPS_MQ订单数据 1" + request);//接收
                                syncMethod<transData> data = new syncMethod<>();
                                data =(syncMethod<transData>) JSON.parseObject(request,new TypeReference<syncMethod<transData>>(){});
                                data.setFrom("CFPS");
                                waybillService.Save(data);
                                apiLogService.updateScuess("CFPS_MQ运单信息");
                            }
                        }
                        //apiLogService.updateScuess("CFPS_MQ");
                    } catch (JMSException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JMSException，请查看日志记录");
                        log.error("CFPS_MQ1 ReceverMsg JMSException syncTradeConsumer"+e.getMessage());
                        //throw new RuntimeException("call ibm mq jms in error,", e);
                    } catch (JsonMappingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonMappingException，请查看日志记录");
                        log.error("CFPS_MQ1 ReceverMsg JsonMappingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (JsonProcessingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonProcessingException，请查看日志记录");
                        log.error("CFPS_MQ1 ReceverMsg JsonProcessingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (Throwable e){
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误，请查看日志记录");
                        log.error("CFPS_MQ1 ReceverMsg Throwable:"+e.getMessage());
                        System.out.println("CFPS_MQ1 ReceverMsg Throwable:"+e.getMessage());
                    }
                }
            });
            connection_cfps1.start();
            //注意需保持此线程不结束监听器才能一直运行
            while (true) {
            }*/
            //注意需保持此线程不结束监听器才能一直运行
        } catch (Exception e) {
//            apiLogService.updateError("MQ");
            e.printStackTrace();
            System.out.println(e.getMessage());
            log.error("10.71.248.34 MQ"+e.getMessage());
        }
    }

    public void ReceverMsgCFPS1() {
        MQQueueConnectionFactory mqQueueConnectionFactory = new MQQueueConnectionFactory();
        Connection connection = null;
        try {
            mqQueueConnectionFactory.setHostName("10.71.248.34");//总线输出的MQ服务器IP
            mqQueueConnectionFactory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
            mqQueueConnectionFactory.setCCSID(1208);
            mqQueueConnectionFactory.setChannel("ESB.ADMIN.SVRCONN");//总线输出的MQ通道
            mqQueueConnectionFactory.setPort(40001);//总线输出的MQ服务器端口
            mqQueueConnectionFactory.setQueueManager("ESB_QM");//总线输出的MQ队列管理器

            connection = mqQueueConnectionFactory.createConnection();

            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("flfj_txd_status_exchange_mq");//分类分级-货邮安检、天信达报文交换MQ通道

            MessageConsumer syncConsumer = session.createConsumer(queue);
            syncConsumer.setMessageListener(new MessageListener() {//设置监听器
                @Override
                public void onMessage(Message message) {
                    try {
                        String request = null;
                        if (message instanceof JMSBytesMessage) {
                            log.info("接收到CFPS_MQ数据JMSBytesMessage 1");
                            JMSBytesMessage jmsBytesMessage = (JMSBytesMessage) message;
                            try {
                                if (null != jmsBytesMessage) {
                                    Long lenght = 2048L;
                                    if (lenght != null && lenght > 0) {
                                        lenght = jmsBytesMessage.getBodyLength();
                                    }
                                    byte[] read = new byte[lenght.intValue()];
                                    jmsBytesMessage.readBytes(read);
                                    request = new String(read, "UTF-8");
                                    log.info("接收到CFPS_MQ数据JMSBytesMessage：" + request);//接收到的消息
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (message instanceof TextMessage) {
                            log.info("接收到CFPS_MQ数据TextMessage 1");
                            TextMessage text = (TextMessage) message;
                            request = text.getText();
                            log.info("接收到CFPS_MQ数据TextMessage" + request);//接收
                        }
                        if(!request.equals(null) && !request.isEmpty()){
                            syncMethod<Object> sync = new syncMethod<>();
                            sync = JsonHelper.getInstance().readValue(request,sync.getClass());
                            if(sync.getMethod().equals("syncDeclaration")){
                                log.info("接收到CFPS_MQ订单数据" + request);//接收
                                syncMethod<transData> data = new syncMethod<>();
                                data =(syncMethod<transData>) JSON.parseObject(request,new TypeReference<syncMethod<transData>>(){});
                                data.setFrom("CFPS");
                                waybillService.Save(data);
                                apiLogService.updateScuess("CFPS_MQ运单信息");
                            }
                        }
                        //apiLogService.updateScuess("CFPS_MQ");
                    } catch (JMSException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JMSException，请查看日志记录");
                        log.error("CFPS_MQ ReceverMsg JMSException syncTradeConsumer"+e.getMessage());
                        //throw new RuntimeException("call ibm mq jms in error,", e);
                    } catch (JsonMappingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonMappingException，请查看日志记录");
                        log.error("CFPS_MQ ReceverMsg JsonMappingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (JsonProcessingException e) {
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误JsonProcessingException，请查看日志记录");
                        log.error("CFPS_MQ ReceverMsg JsonProcessingException syncTradeConsumer"+e.getMessage());
                        e.printStackTrace();
                    } catch (Throwable e){
                        apiLogService.updateError("CFPS_MQ运单信息");
                        runLogService.saveError("CFPS_MQ异常","发生错误，请查看日志记录");
                        log.error("CFPS_MQ ReceverMsg Throwable:"+e.getMessage());
                        System.out.println("CFPS_MQ ReceverMsg Throwable:"+e.getMessage());
                    }
                }
            });
            connection.start();
            while (true) {
            }
            //注意需保持此线程不结束监听器才能一直运行
        } catch (Exception e) {
            log.error("catch MQ ReceverMsg Throwable:"+e.getMessage());
//            apiLogService.updateError("CFPS_MQ");
            e.printStackTrace();
            System.out.println(e.getMessage());
        }
    }



    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("开始执行消息订阅MQ");
        ReceverMsg();
        System.out.println("开始执行消息订阅");
    }
}

